﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.IO;
using System.Configuration;
using System.Diagnostics;

namespace SortFile
{
    class LargeFile
    {
        #region---- Chunk Size ---
        private int _chunkSize = 0;
        private static readonly string _Key_ChunkSize = "ChunkSize";
        private const int _DefaultChunkSize = 10; // MB
        private int ChunkSize
        {
            get {
                var cs = _chunkSize != 0 ? _chunkSize : Convert.ToInt32(ConfigurationManager.AppSettings[_Key_ChunkSize]);
                _chunkSize = cs != 0 ? cs : _DefaultChunkSize;
                return _chunkSize;
            }
        } 
        #endregion

        #region---- Working Folder ---
        private string _workingFolder = null;
        private static readonly string _Date_Part = "_yy_MM_dd_hh_mm_ss_ff";
        private static readonly string _Key_WorkingFolder = "WorkingFolder";
        private static readonly string _DefaultWorkingFolder = "Temp";
        private string WorkingFolder
        {
            get
            {
                if (string.IsNullOrEmpty(_workingFolder))
                {
                    _workingFolder = ConfigurationManager.AppSettings[_Key_WorkingFolder];
                    if (string.IsNullOrEmpty(_workingFolder))
                    {
                        _workingFolder = System.IO.Directory.GetCurrentDirectory();
                        _workingFolder = Path.Combine(_workingFolder, _DefaultWorkingFolder);
                    }
                }
                return _workingFolder;
            }
        }
        #endregion

        #region--- Concurrency ---
        private Object mutex = new object();
        #endregion

        #region---- Sort Chunks ---
        private static readonly string _Pfx_Sort_File = "srt_";
        private static readonly string _Key_Parallel_Sorts = "Parallel_Sorts";
        private const int _DefParallelSorts = 1; // if not 1 then % of sort files
        private int _parallelSorts = 0;
        private int ParallelTracks
        {
            get
            {
                var pt = _parallelSorts != 0 ? _parallelSorts : Convert.ToInt32(ConfigurationManager.AppSettings[_Key_Parallel_Sorts]);
                _parallelSorts = pt != 0 ? pt : _DefParallelSorts;
                return _parallelSorts;
            }
        }

        private List<string> _chunkFiles = new List<string>();
        private int _chunkToSort;
        #endregion

        #region---- Merge Chunks ---
        private static readonly string _Pfx_mrg_File = "m_";
        private static readonly string _Key_Parallel_Merges = "Parallel_Merges";
        private const int _DefParallelMrgs = 1;
        private int _parallelMerges = 0;
        private int ParallelMerges
        {
            get
            {
                var pm = _parallelMerges != 0 ? _parallelMerges : Convert.ToInt32(ConfigurationManager.AppSettings[_Key_Parallel_Merges]);
                _parallelMerges = pm != 0 ? pm : _DefParallelMrgs;
                return _parallelMerges;
            }
        }

        private List<string> _mergeFiles = new List<string>();
        #endregion

        private Stopwatch _stopWatch = new Stopwatch();
       
        #region--- Events ---
        public event FileEvent OnProgress;
        private void InvokeOnprogress(EventInfo info)
        {
            if (OnProgress != null)
            {
                OnProgress(info);
            }
        }
        #endregion

        public void Sort(string filePath)
        {
            #region--- Init variables ---
            // Start stopwatch
            _stopWatch.Restart();
        
            var fileName = Path.GetFileNameWithoutExtension(filePath);
            var ext = Path.GetExtension(filePath);
            _chunkFiles.Clear();

            #endregion

            #region--- Invoke start event ---
            InvokeOnprogress(new EventInfo
            {
                Stage = SortStage.ssStart,
                Message = "Starting the process.."
                                            ,
                ElapsedTime = _stopWatch.ElapsedMilliseconds
            });
            #endregion

            #region--- Split file into chunks ---
            using (var sr = new StreamReader(filePath))
            {
                Directory.CreateDirectory(WorkingFolder);
                while (sr.Peek() >= 0)
                {
                    var chunkFile = MakeFileName(fileName, ext);
                    using (var sw = new StreamWriter(chunkFile))
                    {
                        InvokeOnprogress(new EventInfo { Stage = SortStage.ssMakeChunk, Message = "Making Chunk file : " + chunkFile });
                        Split(sr, sw);
                        _chunkFiles.Add(chunkFile);
                    }
                    InvokeOnprogress(new EventInfo { Stage = SortStage.ssCloseChunk, Message = "Close Chunk file : " + chunkFile });
                }
                InvokeOnprogress(new EventInfo
                {
                    Stage = SortStage.ssFinishedSplit,
                    Message = "Split file to chunks completed."
                                                ,
                    ElapsedTime = _stopWatch.ElapsedMilliseconds
                });             
            }
            #endregion

            #region--- Sort each chunk - Concurrently ---
            _stopWatch.Restart();

            // clear files to merge list
            _mergeFiles.Clear();

            _chunkToSort = _chunkFiles.Count;        
            var numTracks = ParallelTracks;
            if (numTracks != 1)
            {
                double tracks = numTracks / 100.0 * _chunkFiles.Count;
                numTracks = (int)Math.Round(tracks);
            }
            for (var i = 0; i <= numTracks; i++)
            {
                CreateSortTasks(_chunkFiles, null);
                // each sort task will start a track
                // and execute asynchronously 
            }
            #endregion

            // wait till all chunk files are sorted
            while (_chunkToSort > 0) { }

            #region--- Merge sort files ----
            // _mergefiles = list of sorted files
            OnStartMerging();
            string[] filesToMerge = new string[_mergeFiles.Count];
            filesToMerge = _mergeFiles.ToArray();
            var len = filesToMerge.Length;
            while ( len > 1)
            {
                filesToMerge = MergeSortFiles(filesToMerge);
                OnMergedSet(len);
                len = filesToMerge.Length;
            }
            OnFinishedMerging();
            #endregion

        }

        private string[] MergeSortFiles(string[] sortedFiles)
        {
            var newlen = (int)Math.Ceiling(sortedFiles.Length / 2.0);         
            var newFilesToMerge = new string[newlen];
            var file1 = string.Empty;
            var file2 = string.Empty;
            var newIndex = 0;
            Task[] sortTasks = new Task[sortedFiles.Length / 2];
            for (int i = 0; i < sortedFiles.Length; i++)
            {
                if ((i+1) % 2 == 0)
                {
                    file2 = sortedFiles[i];
                    // form merged file name
                    var mergedFile = GetMergeFileName(file2);
                    // add to newFilesToMerge
                    newFilesToMerge[newIndex] = mergedFile;

                    // create mergesort task
                    // need these extra local variables ..
                    // to avoid "closure of lambda" ..
                    // over file1, file2 & mergedFile variables which keep changing :-)
                    var f1 = file1;
                    var f2 = file2;
                    var fo = mergedFile;
                    sortTasks[newIndex] = new Task(() => MergeSort(f1, f2, fo));
                    newIndex++;
                }
                else
                {
                    file1 = sortedFiles[i];
                }
            }
            // add last odd file to list
            if (sortedFiles.Length % 2 != 0)
            {
                newFilesToMerge[newFilesToMerge.Length - 1] = sortedFiles.Last();
            }
            //start merge sort tasks to work in parallel
            foreach (var task in sortTasks)
            {
                task.Start();
            }
            // wait for the tasks to finish
            Task.WaitAll(sortTasks);
            return newFilesToMerge;
        }
        private void MergeSort(string file1, string file2, string outFile)
        {
            //dummy implementation
            lock(mutex)
            {                               
                string line1 = null;
                string line2 = null;
                using (StreamReader sr1 = new StreamReader(file1))
                {
                    using (StreamReader sr2 = new StreamReader(file2))
                    {
                        using (StreamWriter sw = new StreamWriter(outFile))
                        {
                            #region--- merge sort algorith----
                            line1 = sr1.ReadLine();
                            line2 = sr2.ReadLine();
                            while (line1 != null && line2 != null)
                            {                                
                                if (string.Compare(line1, line2) < 0)
                                {
                                    sw.WriteLine(line1);
                                    line1 = sr1.ReadLine();
                                }
                                else
                                {
                                    sw.WriteLine(line2);
                                    line2 = sr2.ReadLine();
                                }
                            }
                            while (line1 != null)
                            {
                                sw.WriteLine(line1);
                                line1 = sr1.ReadLine();
                            }
                            while (line2 != null)
                            {
                                sw.WriteLine(line2);
                                line2 = sr2.ReadLine();
                            }
                            #endregion
                        }
                    }
                }
                File.Delete(file1);
                File.Delete(file2);
            }            
        }
        private string GetMergeFileName(string filename)
        {
            //var dt = DateTime.Now.ToString(_Date_Part);
            var folder = Path.GetDirectoryName(filename);
            var ext = Path.GetExtension(filename);
            var file = Path.GetFileNameWithoutExtension(filename);
            file = _Pfx_mrg_File + file;
            return Path.Combine(folder, file + ext);
        }
        private void CreateSortTasks(List<string> fileList, Task sortTask)
        {
            var file = string.Empty;
            var start = false;
            lock(mutex)
            {
                var rem = Pop(fileList, out file);
                if ((!string.IsNullOrEmpty(file)) && (rem >= 0))
                {
                    if (sortTask == null)
                    {
                        sortTask = new Task(() => SortFile(file, sortTask));
                        start = true;             
                    }
                    else
                    {
                        sortTask = sortTask.ContinueWith(t => SortFile(file, sortTask));
                    }
                    if (rem > 0)
                    {
                        sortTask.ContinueWith(t => CreateSortTasks(fileList, t));
                    }                                 
                }
                if (start) sortTask.Start();
            }
        }
        private void SortFile(string filePath, Task sortTask)
        {
            string[] contents = File.ReadAllLines(filePath);
            // Sort the in-memory array
            Array.Sort(contents);

            var sortedFile = Path.GetFileNameWithoutExtension(filePath);
            var path = Path.GetDirectoryName(filePath);
            var ext = Path.GetExtension(filePath);
            sortedFile = Path.Combine(path, _Pfx_Sort_File + sortedFile + ext);

            File.WriteAllLines(sortedFile, contents);
            File.Delete(filePath);
            // reduce the number of files left to sort
            _chunkToSort--;

            OnSortedChunk(filePath, sortedFile);
            //Console.WriteLine("------ Sort by Task Id = {0}", sortTask.Id);

            //Check if there are any file left for sorting..
            if (_chunkToSort == 0)
            {
                OnFinishedChunkSorting();
            }
        }
        private void OnSortedChunk(string chunkFile, string sortedFile)
        {
            _mergeFiles.Add(sortedFile);
            InvokeOnprogress(new EventInfo
            {
                Stage = SortStage.ssSorting,
                Message = "Sorted file " + chunkFile,
            });
        }
        private void OnFinishedChunkSorting()
        {
            _stopWatch.Stop();
            InvokeOnprogress(new EventInfo
            {
                Stage = SortStage.ssSorted,
                Message = "Finished Sorting chunk files",
                ElapsedTime = _stopWatch.ElapsedMilliseconds
            });
            _stopWatch.Restart();
        }
        private void OnStartMerging()
        {
            InvokeOnprogress(new EventInfo
            {
                Stage = SortStage.ssStartMerge,
                Message = "Start Merging .. sorted chunks ..",              
            });
        }
        private void OnMergedSet(int numOfFiles)
        {
            InvokeOnprogress(new EventInfo
            {
                Stage = SortStage.ssMerging,
                Message = string.Format(".. Merged {0} file ..", numOfFiles),
            });
        }
        private void OnFinishedMerging()
        {
            _stopWatch.Stop();
            InvokeOnprogress(new EventInfo
            {
                Stage = SortStage.ssMerged,
                Message = "Finished Merge Sorting the files",
                ElapsedTime = _stopWatch.ElapsedMilliseconds
            });
        }
        private int Pop(List<string> list, out string file)
        {
            // Concurrency lock is handled by the 
            // calling function.
            // Pop itself is NOT Thread Safe
            file = string.Empty;
            var rem = 0;
            if (list.Count > 0)
            {
                file = list[0];
                list.RemoveAt(0);
                rem = list.Count;
            }
            return rem;
        }

        private string MakeFileName(string name, string ext)
        {
            var dt = DateTime.Now.ToString("_yy_MM_dd_hh_mm_ss_ff");
            return Path.Combine(WorkingFolder, name + dt + ext);
        }
        private void Split(StreamReader sr, StreamWriter sw)
        {
            var size = ChunkSize * 1e6;
            bool lessThanSize = true;
            while (lessThanSize && sr.Peek() >= 0)
            {
                long chunkSize = sw.BaseStream.Length;
                var ln = sr.ReadLine();
                sw.WriteLine(ln);
                lessThanSize = chunkSize < size;
                InvokeOnprogress(new EventInfo { Stage = SortStage.ssWriteLine, Message = "Write file size : " + chunkSize.ToString() });
            }

        }
    }

    public delegate void FileEvent(EventInfo info);
    public struct EventInfo
    {
        public SortStage Stage { get; set; }
        public string Message { get; set; }

        public long ElapsedTime { get; set; }
    }
    public enum SortStage { ssNone, ssStart, ssMakeChunk, ssWriteLine, ssCloseChunk, ssFinishedSplit
            , ssStartSort, ssSorting, ssSorted, ssStartMerge, ssMerging, ssMerged}

}
